package com.ksyun.seckill.service.impl.V7;


import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.ksyun.seckill.constants.RedisKey;
import com.ksyun.seckill.constants.SecKillImplVer;
import com.ksyun.seckill.dto.CampaignCacheInfoDto;
import com.ksyun.seckill.dto.PurchaseOrderInfo;
import com.ksyun.seckill.entity.Campaign;
import com.ksyun.seckill.entity.PurchaseOrder;
import com.ksyun.seckill.keygen.RedisIdWorker;
import com.ksyun.seckill.service.BizService;
import com.ksyun.seckill.service.CampaignService;
import com.ksyun.seckill.service.OrderService;
import com.ksyun.seckill.service.SkService;
import com.ksyun.seckill.util.exception.ServiceException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.DigestUtils;
import org.springframework.util.ObjectUtils;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

@Service("SeckillService")
@Slf4j
public class DefaultSeckillServiceImplV7 implements SkService {

    @Autowired
    private BizService bizService;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private CampaignService campaignService;

    @Autowired
    private OrderService orderService;


    @Value("${redis.mq.queueName}")
    private String queueName ;

    @Value("${redis.mq.group1}")
    private String group1;

    @Value("${redis.mq.consumer1}")
    private String consumer1;

    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    @Autowired
    private RedisIdWorker redisIdWorker;

    private CampaignCacheInfoDto cache;

    //当前类的代理对象
    private DefaultSeckillServiceImplV7 proxy;


    /**
     * 秒杀业务脚本
     */
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

    /**
     * 创建消费者组和消息队列的脚本
     *
     */
    private static final DefaultRedisScript<Long> STREAM_GROUP_INIT_SCRIPT;

    static {
        STREAM_GROUP_INIT_SCRIPT = new DefaultRedisScript<>();
        STREAM_GROUP_INIT_SCRIPT.setLocation(new ClassPathResource("initStream.lua"));
        STREAM_GROUP_INIT_SCRIPT.setResultType(Long.class);

        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    @PostConstruct
    private void init() {
        stringRedisTemplate.execute(STREAM_GROUP_INIT_SCRIPT,Collections.emptyList(),queueName,group1);
        //开启线程
        SECKILL_ORDER_EXECUTOR.submit(new OrderHandler());
    }

    //线程任务
    private class OrderHandler implements Runnable {
        @Override
        public void run() {
            while (true) {
                if(proxy==null){
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    continue;
                }
                try {
                    //1、从消息队列获取订单 xreadgroup group g1 c1 count 1 block 2000 streams stream.orders >
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from(group1, consumer1),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create(queueName, ReadOffset.lastConsumed())
                    );
                    log.info(list.toString());
                    //2、判断消息是否获取成功
                    if (list == null || list.isEmpty()) {
                        //2、1 如果获取失败，说明没有消息，继续下一次循环
                        continue;
                    }
                    //3、 如果获取成功，进行处理订单的逻辑
                    //3、1 解析消息中的订单信息
                    MapRecord<String, Object, Object> record = list.get(0);
                    //消息id、key、value
                    Map<Object, Object> value = record.getValue();
                    PurchaseOrder order = BeanUtil.fillBeanWithMap(value, new PurchaseOrder(), true);
                    handlerPurchaseOrder(order);
                    //4、ack确认，确认消息 xack stream.orders g1 id
                    stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
                } catch (Exception e) {
                    log.error("处理订单时异常,{}", e.getMessage());
                    //异常，说明消息未被确认
                    //1、去pendinglist中去消费并且确认消息
                    handlePendingList();
                }
            }
        }

        /**
         * 去pendinglist中去消费并且确认消息
         */
        private void handlePendingList() {
            while (true) {
                try {
                    //1、从pendinglist中 xreadgroup group g1 c1 count 1 block 2000 streams stream.orders 0
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from(group1, consumer1),
                            StreamReadOptions.empty().count(1),
                            StreamOffset.create(queueName, ReadOffset.from("0"))
                    );
                    //2、判断消息是否获取成功
                    if (list == null || list.isEmpty()) {
                        //2、1 如果获取失败，说明pendinglist中没有消息，结束循环
                        break;
                    }
                    //3、 如果获取成功，进行处理订单的逻辑
                    //3、1 解析消息中的订单信息
                    MapRecord<String, Object, Object> record = list.get(0);
                    //消息id、key、value
                    Map<Object, Object> value = record.getValue();
                    PurchaseOrder order = BeanUtil.fillBeanWithMap(value, new PurchaseOrder(), true);
                    handlerPurchaseOrder(order);
                    //4、ack确认，确认消息 xack stream.orders g1 id
                    stringRedisTemplate.opsForStream().acknowledge(queueName, group1, record.getId());

                } catch (Exception e) {
                    //异常，说明在处理pendinglist中出现异常，继续循环即可;
                    try {
                        //休眠一下，防止太频繁
                        Thread.sleep(20);
                    } catch (InterruptedException ex) {
                        ex.printStackTrace();
                    }
                }
            }
        }
    }

    /**
     * 线程任务处理订单任务
     *
     * @param order
     */
    private void handlerPurchaseOrder(PurchaseOrder order) {
        /**
         * 这里应该不会出现并发问题，因为在lua中已经判断好了，不需要再获取锁了
         */
        proxy.createOrder(order);
    }

    /**
     * 创建订单
     *
     * @param order
     */
    @Transactional(rollbackFor = Exception.class)
    public void createOrder(PurchaseOrder order) {
        String userPhone = order.getUserPhone();
        Long campaignId = order.getCampaignId();
        //查询订单
        QueryWrapper<PurchaseOrder> wrapper = new QueryWrapper<>();
        wrapper.eq("campaign_id", campaignId).eq("user_phone", userPhone);
        Integer count = orderService.getBaseMapper().selectCount(wrapper);
        if (count > 0) {
            //不太可能重复
            log.error("{}用户已经购买过一次{}!", userPhone, campaignId);
        }
        //扣减库存
        boolean success = campaignService.update().setSql("current_inventory=current_inventory-1").
                eq("id", campaignId).gt("current_inventory", 0).update();
        if (!success) {
            //扣减失败
            log.info("库存不足");
        }
        orderService.save(order);
    }


    /**
     * 检验秒杀活动的合法性，如活动是不存在，活动是否允许参与秒杀等.
     *
     * @param campaignId 活动id.
     */
    @Override
    public void validateCampaign(Long campaignId) {
        //1、id是否合法
        Map<Object, Object> map = stringRedisTemplate.opsForHash().
                entries(RedisKey.CAMPAIGN_INFO.formatKey(campaignId));
        if (map.isEmpty()) {
            throw new ServiceException("活动不存在!");
        }
        //2、查询到的活动的状态是否正确
        cache = BeanUtil.fillBeanWithMap(map, new CampaignCacheInfoDto(), false);
        //2、2 活动未结束
        if (LocalDateTime.now().isBefore(cache.getBeginTime())) {
            throw new ServiceException("活动未开始!");
        }
        if (LocalDateTime.now().isAfter(cache.getEndTime())) {
            throw new ServiceException("活动已结束!");
        }
        //2、3 活动的储存要够
        if (cache.getCurrentInventory() <= 0) {
            throw new ServiceException("活动没有库存了！");
        }
    }


    /**
     * 指定活动，用户是否已下过单，同一个手机号，同一个活动，只能参与一次.
     *
     * @param campaignId 活动id.
     * @param phone      手机号.
     * @return
     */
    @Override
    public Boolean haveOrder(Long campaignId, String phone) {
        //从redis中判断是否已经下过单
        Boolean ismember = stringRedisTemplate.opsForSet().
                isMember(RedisKey.HAVE_ORDER_USER_LIST.formatKey(campaignId), phone);
        return ismember;
    }


    /**
     * 秒杀.
     * TODO 在活动上线的那一刻就将其存入至redis中。
     *
     * @param campaignId 活动id.
     * @param phone      手机号.
     * @param path       秒杀URL路径.
     */
    @Override
    public void secKill(Long campaignId, String phone, String path) {
        //1、首先验证path是否合法
        boolean isValidate = checkSecKillPath(campaignId, phone, path);
        if (!isValidate) {
            throw new ServiceException("秒杀路径不合法!");
        }
        //2、判断活动id是否合法,活动是否能够参与秒杀等等
        validateCampaign(campaignId);

        //2、1 封装订单信息
        PurchaseOrderInfo order = encapsulateOrder(campaignId, phone);
        //3、判断库存是否充足，以及一人一单，扣减库存，保存用户，以及将下单的订单信息发送到消息队列中去的业务逻辑放在lua脚本中去进行
        //3、1 执行lua脚本
        Long result = stringRedisTemplate.execute(SECKILL_SCRIPT, Collections.emptyList(),
                campaignId.toString(), phone, order.getId(), order.getCampaignName(), order.getStatus(), order.getUpdateTime(),
                order.getExpirationTime(), order.getFriendCode(), order.getRemarks(), order.getCreateTime(),queueName);
        //3、2 判断结果是否为0
        int ret = result.intValue();
        if (ret != 0) {
            String msg = ret == 1 ? "库存不足!" : "不能重复下单!";
            throw new ServiceException(msg);
        }
        proxy= (DefaultSeckillServiceImplV7) AopContext.currentProxy();

    }


    /**
     * 根据活动id和phone封装订单信息，全部是string类型，传入lua
     *
     * @param campaignId
     * @param phone
     * @return
     */
    private PurchaseOrderInfo encapsulateOrder(Long campaignId, String phone) {
        PurchaseOrderInfo order = new PurchaseOrderInfo();
        //1、生成order的id
        long orderId = redisIdWorker.nextId("order");
        order.setId(String.valueOf(orderId));
        //2、phone
        order.setUserPhone(phone);
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        //3、campaign
        order.setCampaignId(String.valueOf(campaignId));
        //4、其他
        order.setCampaignName(cache.getName());
        order.setStatus("1");
        order.setUpdateTime(LocalDateTime.now().format(formatter));
        order.setExpirationTime(LocalDateTime.now().plusMonths(6l).format(formatter));
        order.setFriendCode(bizService.generateFriendCode());
        order.setRemarks(cache.getDescr());
        order.setCreateTime(LocalDateTime.now().format(formatter));
        return order;
    }


    /**
     * 服务实现版本.
     *
     * @return
     */
    @Override
    public SecKillImplVer getSecKillImplVer() {
        return SecKillImplVer.V7;
    }

    /**
     * 生成秒杀URL路径.
     *
     * @param campaignId 活动id.
     * @param phone      手机号.
     * @param verifyCode 验证码
     * @return
     */
    @Override
    public String generateSecKillPath(Long campaignId, String phone, String verifyCode) {
        /**
         * 思路：
         *  1、可以将接收到的campaignId和phone进行md5加密，将加密后的字符串存入到redis中，然后返回给用户
         *
         *  2、在真正进行秒杀业务前进行检查url是否正确
         *
         *  状态：
         *      1、只有上线的活动才能进行查询活动
         *      1:新建  2:已上线　3.已结束 4.永久下线
         *
         */
        //查询秒杀商品的信息
        Campaign campaign = campaignService.getById(campaignId);
        if (ObjectUtils.isEmpty(campaign)) {
            throw new ServiceException("不存在对应的活动！");
        }
        if (campaign.getStatus() != 2) {
            throw new ServiceException("活动未上线！");
        }
        String salt = "random_salt_" + campaign.getId() + phone;
        String md5 = DigestUtils.md5DigestAsHex(salt.getBytes());
        //存储至redis中
        stringRedisTemplate.opsForValue().set(RedisKey.SEC_KILL_PATH.formatKey(phone, campaign.getId())
                , md5, RedisKey.SEC_KILL_PATH_TTL, TimeUnit.MINUTES);

        return md5;
    }

    /**
     * 检查秒杀URL路径合法性.
     * 在真正执行秒杀前需要检查URL是否正确，避免通过非法方式进入秒杀页面
     *
     * @param campaignId 活动id.
     * @param phone      手机号.
     * @param path       秒杀URL路径.
     * @return
     */
    @Override
    public boolean checkSecKillPath(Long campaignId, String phone, String path) {
        if (path == null) {
            return false;
        }
        String pathOld = stringRedisTemplate.opsForValue().
                get(RedisKey.SEC_KILL_PATH.formatKey(phone, campaignId));
        return path.equals(pathOld);
    }


}
